package com.xxxx;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

import java.util.Iterator;

public class Hello10FlinkProcessWindowFunction {
    public static void main(String[] args) throws Exception {
        //1.初始化环境
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        environment.setParallelism(1);
        //2.读取数据源
        DataStreamSource<String> source = environment.socketTextStream("192.168.88.101", 18880);
        source.map(word -> {
            System.err.println(word + "--" + System.currentTimeMillis());
            return Tuple2.of(word, 1);
        }).returns(Types.TUPLE(Types.STRING, Types.INT)).keyBy(0).countWindow(5).process(new ProcessWindowFunction<Tuple2<String, Integer>, String, Tuple, GlobalWindow>() {
            @Override
            public void process(Tuple tuple, Context context, Iterable<Tuple2<String, Integer>> iterable, Collector<String> collector) throws Exception {
                String key = null;
                int sum = 0;
                //开始迭代数据
                Iterator<Tuple2<String, Integer>> iterator = iterable.iterator();
                for (Tuple2<String, Integer> tuple2 : iterable) {
                    key = tuple2.f0;
                    sum += tuple2.f1;
                }
                //收集数据
                collector.collect(key + "--" + sum);
            }
        }).print();
        //3.需要手动触发流式计算的执行效果
        environment.execute("Hello10FlinkSessionWindow" + System.currentTimeMillis());
    }
}